package com.kl.ordermq;

import com.alibaba.fastjson.JSONObject;
import com.kl.entity.SeckillCoupon;
import com.kl.entity.SeckillOrder;
import com.kl.service.ISeckillCouponService;
import com.kl.service.ISeckillOrderService;
import com.kl.util.IdWorker;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.Date;
import java.util.List;
import java.util.Map;

@Component
@RocketMQMessageListener(topic = "sync_topic",consumerGroup = "seckill_group2")
public class OrderStateMQ implements RocketMQListener, RocketMQPushConsumerLifecycleListener {

    @Resource
    private RedisTemplate redisTemplate;

    @Resource
    private ISeckillOrderService iSeckillOrderService;

    @Override
    public void onMessage(Object o) {}

    @Override
    public void prepareStart(DefaultMQPushConsumer defaultMQPushConsumer) {

        defaultMQPushConsumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                //获取的消息
                MessageExt messageExt = list.get(0);
                //消息的重试次数
                int times = messageExt.getReconsumeTimes();
                try {
                    //消息的id
                    String msgId = messageExt.getMsgId();
                    //消息幂等性
                    Boolean S = redisTemplate.boundHashOps("OR_MQmessage").hasKey(msgId);
                    if (!S){
                        //消息内容
                        byte[] body = messageExt.getBody();
                        //转成string
                        String message = new String(body);
                        System.out.println(message);


                        //消费成功放入redis
                        redisTemplate.boundHashOps("OR_MQmessage").put(msgId,messageExt);

                        //业务逻辑 根据id查询订单查询
                        long id = Long.parseLong(message);
                        SeckillOrder order = iSeckillOrderService.getById(id);
                        Long seckillCouponId = order.getSeckillCouponId();
                        if (order.getPaymentStatus() == 0){

                            order.setPaymentStatus(3);//交易关闭
                            order.setId(id);
                            order.setClosingTime(new Date());
                            order.updateById();


                            String num = (String)redisTemplate.boundValueOps(seckillCouponId).get();

                            if (Integer.parseInt(num) < 0 || StringUtils.isBlank(num)){
                               redisTemplate.boundValueOps(seckillCouponId).set(0);
                           }
                            Long increment = redisTemplate.boundValueOps(seckillCouponId).increment();
                            //同步到redis
                            SeckillCoupon coupon = (SeckillCoupon)redisTemplate.boundHashOps("kl_SeckillCoupon").get(seckillCouponId);
                            coupon.setCouponCount(increment.intValue());
                            redisTemplate.boundHashOps("kl_SeckillCoupon").put(seckillCouponId,coupon);
                        }


                    }
                }catch (Exception e){
                    //消费失败 判断重试次数 3次之后返回消费成功
                    e.printStackTrace();
                    if (times > 3){
                        //放入自定义死信队列
                        redisTemplate.boundHashOps("OR_DeadLetter").put(messageExt.getMsgId(),messageExt);
                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                    }
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }
                //消费成功
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
    }
}
